package com.inns.position.controller;

import com.alibaba.fastjson.JSON;
import com.inns.position.exceptioin.InnsException;
import com.inns.position.exceptioin.InnsExceptionEnum;
import com.inns.position.hbase.HbaseConsts;
import com.inns.position.hbase.StatType;
import com.inns.position.hbase.config.properties.InnsProperties;
import com.inns.position.hbase.service.IHBaseService;
import com.inns.position.hbase.service.StatisticsService;
import com.inns.position.hbase.util.HbaseUtil;
import com.inns.position.kafka.bean.LocationInfo;
import com.inns.position.redis.service.RealTimeDataService;
import com.inns.position.utils.DateUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.Random;

/**
 * Created by phoenix on 2018/4/7.
 */
@RestController
public class TestController {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RealTimeDataService realTimeDataService;

    @Resource
    private StatisticsService statisticsService;

    @Resource
    private IHBaseService ihBaseService;

    @GetMapping("ok")
    public String ok() {
        InnsProperties props = InnsProperties.instance();
        System.out.println(props.getAppId());
        System.out.println(props.getMinTime());
        System.out.println(props.getOfflineTime());
        System.out.println(props.getRealTimeOnlineTime());
        return "ok";
    }

    @GetMapping("exception")
    public Object exception(Integer type) {
        switch (type) {
            case 1:
                return json();
            case 2:
                throw new InnsException(InnsExceptionEnum.USER_NOT_FIND);
            default:
                throw new RuntimeException("test other error!");
        }
    }

    @GetMapping("json")
    public Object json() {
        LocationInfo locationInfo = new LocationInfo();
        locationInfo.setIMSI("001");
        locationInfo.setBuildId("b1");
        locationInfo.setFloorId("f1");
        locationInfo.setPoiId(123);
        locationInfo.setTimestamp(System.currentTimeMillis());
        return locationInfo;
    }

    @GetMapping("list")
    public Object list(List<String> list) {
        return list;
    }


    @GetMapping("redisAdd")
    public String redisAdd() {
        redisTemplate.opsForValue().set("json", json());
        realTimeDataService.updateUserData((LocationInfo) json());
        return "add success";
    }

    @GetMapping("redisGet")
    public Object redisGet() {
        Object json = redisTemplate.opsForValue().get("json");
        return json;
    }

    @GetMapping("producer")
    public String producer() {
        ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send("test", "key", "testValue");
        try {
            SendResult<String, String> sendResult = send.get();
            logger.warn(sendResult.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "producer success";
    }

    @GetMapping("producerTestInfo")
    public String producerTestInfo(@RequestParam(required = false, defaultValue = "1") int num,
                                   @RequestParam(defaultValue = HbaseConsts.TableName_Original) String topic) {
        Random random = new Random();
        for (int i = 0; i < num; i++) {
            LocationInfo locationInfo = new LocationInfo();
            locationInfo.setIMSI("Imsi" + DateUtil.formatCurrentDate("yyyyMMddHHmmss") + "_" + i);
            locationInfo.setBuildId("b" + random.nextInt(10));
            locationInfo.setFloorId("f" + random.nextInt(30));
            locationInfo.setPoiId(random.nextInt(5));
            locationInfo.setTimestamp(System.currentTimeMillis());
            locationInfo.setX(random.nextInt(100));
            locationInfo.setY(random.nextInt(100));
            String jsonString = JSON.toJSONString(locationInfo);
            ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, "key", jsonString);
            try {
                SendResult<String, String> sendResult = send.get();
//                logger.warn(sendResult.toString());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return "producer success";
    }

    @GetMapping("statistics")
    public String statistics(@RequestParam(required = false, defaultValue = "1") int stat_type,
                             @RequestParam(required = false, defaultValue = "4") int m,
                             @RequestParam(required = false, defaultValue = "25") int d,
                             @RequestParam(required = false) int h,
                             @RequestParam(required = false, defaultValue = "0") int minute,
                             @RequestParam int i) {
        long curTime = DateUtil.newDate(2018, m, d, h, minute).getTime();
        StatType statType = StatType.valueOf(stat_type);
        long startTime = statType.getStartTime(curTime);
        long endTime = statType.getEndTime(curTime);
        switch (i) {
            case 331:
                statisticsService.statistics331(statType, startTime, endTime);
                break;
            case 332:
                statisticsService.statistics332(statType, startTime, endTime);
                break;
            case 333:
                statisticsService.statistics333(statType, startTime, endTime);
                break;
            case 334:
                statisticsService.statistics334(statType, startTime, endTime);
                break;
            case 3341:
                statisticsService.statistics334_poi(statType, startTime, endTime);
                break;
            case 335:
                statisticsService.statistics335(statType, startTime, endTime);
                break;
            case 3351:
                statisticsService.statistics335_poi(statType, startTime, endTime);
                break;
            default:
                return "接口类型错误！i=" + i;
        }
        logger.info(DateUtil.format(startTime, "MM:dd HH:mm") + "--" + DateUtil.format(endTime, "MM:dd HH:mm"));
        return "success";
    }

    @GetMapping("statistics2")
    public String statistics2(@RequestParam(required = false, defaultValue = "22") int d) {
        int m = 6;
        int dayNum = 4;
        // 循环天
        for (int i = d; i < d + dayNum; i++) {
            int type = StatType.DAY.getType();
            statistics(type, m, i, 1, 0, 331);
            logger.info("月=" + m + ",日=" + d);
            statistics(type, m, i, 1, 0, 332);
            logger.info("月=" + m + ",日=" + d);
            statistics(type, m, i, 1, 0, 333);
            logger.info("月=" + m + ",日=" + d);
            statistics(type, m, i, 1, 0, 334);
            logger.info("月=" + m + ",日=" + d);
            statistics(type, m, i, 1, 0, 3341);
            logger.info("月=" + m + ",日=" + d);
        }
        logger.info("1======================================================");
        // 循环天
        for (int i = d; i <= d + dayNum; i++) {
            int day = i - 1;
            // 循环小时
            for (int j = 1; j <= 24; j++) {
                int type = StatType.HOUR.getType();
                statistics(type, m, day, j, 0, 331);
                statistics(type, m, day, j, 0, 332);
                logger.info("月=" + m + ",日=" + d + ",时=" + j);
            }
        }
        logger.info("2======================================================");
        // 循环天
        for (int i = d; i <= d + dayNum; i++) {
            int day = i - 1;
            // 循环小时
            for (int j = 0; j < 24; j++) {
                // 循环分钟
                for (int k = 1; k <= 60; k += 5) {
                    int type = StatType.MINUTE.getType();
                    statistics(type, m, day, j, k, 331);
                    statistics(type, m, day, j, k, 332);
                    logger.info("月=" + m + ",日=" + d + ",时=" + j + ",分=" + k);
                }
            }
        }
        logger.info("3======================================================");
        return "statistics2 over";
    }

    @GetMapping("fixData")
    public String fixData() throws IOException {
        Connection connection = HbaseUtil.getConnection();
        TableName tableName = TableName.valueOf(HbaseConsts.TableName_Original);
        Table table = connection.getTable(tableName);

        long startTime = StatType.DAY.getStartTime(DateUtil.newDate(2018, 6, 22, 12).getTime());
        long endTime = StatType.DAY.getStartTime(DateUtil.newDate(2018, 6, 26, 12).getTime());

        int increaseTime = 10 * 60 * 1000;
        long endTime1 = startTime + increaseTime;
        while (endTime1 <= endTime) {
            logger.warn("aaa startTime=" + startTime + ", endTime=" + endTime1);
            Scan scan = new Scan();
            scan.setTimeRange(startTime, endTime1);
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                List<Cell> rowCells = result.listCells();
                LocationInfo locationInfo = new LocationInfo();
                for (Cell cell : rowCells) {
                    String qualify = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                    String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
//                    System.out.println(family + " " + qualify + " " + value);
                    if (qualify.equals("IMSI")) {
                        locationInfo.setIMSI(value);
                    } else if (qualify.equals("buildId")) {
                        locationInfo.setBuildId(value);
                    } else if (qualify.equals("floorId")) {
                        locationInfo.setFloorId(value);
                    } else if (qualify.equals("poiId")) {
                        locationInfo.setPoiId(Integer.parseInt(value));
                    } else if (qualify.equals("x")) {
                        locationInfo.setX(Float.parseFloat(value));
                    } else if (qualify.equals("y")) {
                        locationInfo.setY(Float.parseFloat(value));
                    }
                }
                locationInfo.setTimestamp(rowCells.get(0).getTimestamp());
                System.out.println(locationInfo.toString());
                try {
                    realTimeDataService.updateUserData(locationInfo);
                    realTimeDataService.updateBuildData(locationInfo);
                    realTimeDataService.updateUserPoiData(locationInfo);
                } catch (Exception e) {
                    logger.error("redis saveInfo error: ", e);
                }
            }
            startTime += increaseTime;
            endTime1 = startTime + increaseTime;
        }
        return "over";
    }
}
